package com.mallcai.bigdata.ladon.service.peekdata.impl;

import com.alibaba.excel.EasyExcelFactory;
import com.alibaba.excel.context.AnalysisContext;
import com.alibaba.excel.event.AnalysisEventListener;
import com.alibaba.excel.metadata.Sheet;
import com.cake.bigdata.starter.exception.category.BusinessException;
import com.google.common.base.Preconditions;
import com.google.common.collect.Sets;
import com.mallcai.bigdata.ladon.dpl.entity.peekdata.ImportRecord;
import com.mallcai.bigdata.ladon.service.common.IDatasourceService;
import com.mallcai.bigdata.ladon.service.peekdata.IDataImportService;
import com.mallcai.bigdata.ladon.service.peekdata.IImportRecordService;
import com.mallcai.bigdata.ladon.utils.CollectionUtils;
import com.mallcai.bigdata.ladon.utils.ExceptionUtils;
import com.mallcai.bigdata.ladon.vo.BatchUpdateParam;
import com.mallcai.bigdata.ladon.vo.ColumnVo;
import com.mallcai.bigdata.ladon.vo.DataVo;
import com.mallcai.bigdata.ladon.vo.QueryParam;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.poi.hssf.usermodel.HSSFDateUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.core.ParameterizedPreparedStatementSetter;
import org.springframework.stereotype.Service;
import org.springframework.web.multipart.MultipartFile;

import java.io.IOException;
import java.io.InputStream;
import java.math.BigDecimal;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.sql.Types;
import java.text.MessageFormat;
import java.util.*;
import java.util.function.Function;
import java.util.stream.Collectors;

import static com.google.common.base.Preconditions.checkNotNull;

/**
 * @author feixy
 * @version 1.0
 * @date 2019-07-12 16:27
 */
@Slf4j
@Service
public class DataImportService implements IDataImportService {

    private static final String DELETE_CURRENT_FORMAT = "delete from %s where %s = %d ";

    private static final String ARCHIVED_ALL_FORMAT = "update %s set %s = 2 where %s = 0 "; // 覆盖删除

    private static final String ARCHIVED_CURRENT_FORMAT = "update %s set %s = 1 where %s = %s "; // 归档指定数据

    public static final String INNER_FIELD_IMPORT_ID = "ladon_import_id";

    private static final String INNER_FIELD_ARCHIVED = "ladon_archived";

    private static final Set<String> INNER_FIELDS = Sets.newHashSet(INNER_FIELD_IMPORT_ID, INNER_FIELD_ARCHIVED);

    @Autowired
    private IImportRecordService iImportRecordService;

    @Autowired
    private IDatasourceService iDatasourceService;

    @Override
    public void doImport(ImportRecord importRecord, MultipartFile file) throws IOException {
        Preconditions.checkNotNull(file);
        List<ColumnVo> columnVoList = iDatasourceService.getTableColumns(importRecord.getDatasourceId(), importRecord.getTableName());
        checkTableColumn(columnVoList);
        iImportRecordService.add(importRecord);

        try(InputStream inputStream = file.getInputStream()){
            EasyExcelFactory.getReader(inputStream,  new ExcelListener(importRecord,columnVoList)).read(new Sheet(1,0));
            iImportRecordService.success(importRecord.getId());
        }catch (Exception e){
            cleanCurrentData(importRecord);
            iImportRecordService.fail(importRecord.getId(), ExceptionUtils.getSimpleMessage(e));
            throw e;
        }
    }

    private void checkTableColumn(List<ColumnVo> columnVoList) {
        Set<String> columns = CollectionUtils.toSet(columnVoList, ColumnVo::getName);
        List<String> missedFields = INNER_FIELDS.stream()
                .filter(f -> !columns.contains(f)).collect(Collectors.toList());
        if (CollectionUtils.isNotEmpty(missedFields)){
            throw new BusinessException(String.format("数据库表中不存在以下系统保留字段[%s],请修改", CollectionUtils.join(missedFields,Function.identity())));
        }
    }

    /**
     * 删除本次已经导入的数据
     * @param importRecord
     */
    private void cleanCurrentData(ImportRecord importRecord) {
        QueryParam param = new QueryParam();
        param.setRawSql(String.format(DELETE_CURRENT_FORMAT, importRecord.getTableName(), INNER_FIELD_IMPORT_ID,importRecord.getId()));
        iDatasourceService.execute(importRecord.getDatasourceId(), param);
    }

    /**
     * 逻辑删除
     * @param importRecord
     */
    private void archivedAllData(ImportRecord importRecord) {
        QueryParam param = new QueryParam();
        param.setRawSql(String.format(ARCHIVED_ALL_FORMAT, importRecord.getTableName(), INNER_FIELD_ARCHIVED, INNER_FIELD_ARCHIVED));
        iDatasourceService.execute(importRecord.getDatasourceId(), param);
    }

    private class ExcelListener extends AnalysisEventListener<List<String>> {

        private List<String> header = null;

        private List<List<String>> rowList = new LinkedList<>();

        private final ImportRecord importRecord;
        private final List<ColumnVo> columnVoList;

        public ExcelListener(ImportRecord importRecord, List<ColumnVo> columnVoList) {
            this.importRecord = importRecord;
            this.columnVoList = columnVoList;
        }

        @Override
        public void invoke(List<String> object, AnalysisContext context) {
            if(context.getCurrentRowNum() == 0){
                Function<String,String> function = String::toLowerCase;
                header = CollectionUtils.map(object,  function.andThen(String::trim));
            } else {
                if(CollectionUtils.isNotEmpty(object)){
                    boolean hasOneValue = object.stream().anyMatch(StringUtils::isNotEmpty);
                    if(hasOneValue){
                        rowList.add(object);
                    }
                }
            }
        }

        @Override
        public void doAfterAllAnalysed(AnalysisContext context) {
            doCheck(this.columnVoList);

            Map<String,ColumnVo> columnVoMap = CollectionUtils.toMap(columnVoList, columnVo -> columnVo.getName().toLowerCase(), Function.identity());
            // 先删除
            if(1 == importRecord.getOverwrited()){
                archivedAllData(importRecord);
            }

            List<String> allHeaders = new ArrayList<>();
            allHeaders.addAll(this.header);
            allHeaders.addAll(INNER_FIELDS);

            Map<String, String> innerParams = getInnerParams();

            String fieldPlaceHolder = CollectionUtils.join(allHeaders, Function.identity());
            String valuePlaceHolder = CollectionUtils.join(CollectionUtils.map(allHeaders,(p)-> "?"), Function.identity());

            int headerNum = allHeaders.size();
            ParameterizedPreparedStatementSetter<List<String>> statementSetter = (PreparedStatement ps, List<String> strList) -> {
                for (int i = 0; i < headerNum; i++) {
                    String colName = allHeaders.get(i);
                    ColumnVo columnVo = columnVoMap.get(colName);
                    String value;
                    if (i < header.size() && i < strList.size()){
                        value = strList.get(i);
                    } else {
                        value = innerParams.get(colName);
                    }
                    try{
                        setPreparedStatementValue(ps, i+1, columnVo, value);
                    }catch (Exception e){
                        throw new BusinessException(String.format("数据类型错误,列:[%s],值:[%s],期望类型:[%s]",colName, value, columnVo.getOriginalType()));
                    }
                }
            };

            BatchUpdateParam<List<String>> batchUpdateParam = new BatchUpdateParam<>();
            String sql = MessageFormat.format("insert into {0} ({1}) values ({2})", importRecord.getTableName(),fieldPlaceHolder, valuePlaceHolder);
            log.info("insert sql:{}", sql);
            batchUpdateParam.setSql(sql);
            batchUpdateParam.setBatchSize(3);
            batchUpdateParam.setParameterizedSetter(statementSetter);
            batchUpdateParam.setParamList(this.rowList);

            iDatasourceService.batchUpdate(importRecord.getDatasourceId(), batchUpdateParam);

        }

        private Map<String, String> getInnerParams() {
            Map<String,String> innerFieldParams = new HashMap<>();
            innerFieldParams.put(INNER_FIELD_IMPORT_ID, String.valueOf(importRecord.getId()));
            innerFieldParams.put(INNER_FIELD_ARCHIVED, String.valueOf(0));
            return innerFieldParams;
        }

        private void doCheck(List<ColumnVo> tableColumns){
            if(CollectionUtils.isEmpty(header) || CollectionUtils.isEmpty(rowList)){
                throw new BusinessException("文件内容为空,请检查");
            }
            Set<String> columns = CollectionUtils.toSet(tableColumns, ColumnVo::getName);
            for (String h : header){
                if (INNER_FIELDS.contains(h)){
                    throw new BusinessException(String.format("文件中的字段[%s]为系统保留字段,请修改", h));
                }
                if (!columns.contains(h)){
                    throw new BusinessException(String.format("文件中的字段[%s]在数据库表中不存在,请修改", h));
                }
            }
        }
    }


    private static void setPreparedStatementValue(PreparedStatement ps, int index, ColumnVo columnVo, String value) throws SQLException {
        String originDataType = columnVo.getOriginalType();
        if ("TINYINT".equalsIgnoreCase(originDataType)){
            if (StringUtils.isNotEmpty(value)){
                ps.setByte(index, Byte.valueOf(value.trim()));
            } else {
                ps.setNull(index, Types.TINYINT);
            }
        } else if("SMALLINT".equalsIgnoreCase(originDataType)){
            if (StringUtils.isNotEmpty(value)){
                ps.setShort(index, Short.valueOf(value.trim()));
            } else {
                ps.setNull(index, Types.SMALLINT);
            }
        } else if("MEDIUMINT".equalsIgnoreCase(originDataType) || "INT".equalsIgnoreCase(originDataType) || "INTEGER".equalsIgnoreCase(originDataType)){
            if (StringUtils.isNotEmpty(value)){
                ps.setInt(index, Short.valueOf(value.trim()));
            } else {
                ps.setNull(index, Types.INTEGER);
            }
        } else if("BIGINT".equalsIgnoreCase(originDataType)){
            if (StringUtils.isNotEmpty(value)){
                ps.setLong(index, Long.valueOf(value.trim()));
            } else {
                ps.setNull(index, Types.BIGINT);
            }
        } else if( "VARCHAR".equalsIgnoreCase(originDataType) || "CHAR".equalsIgnoreCase(originDataType)){
            if (StringUtils.isNotEmpty(value)){
                ps.setString(index,value.trim());
            } else {
                ps.setNull(index, Types.VARCHAR);
            }
        } else if ("DOUBLE".equalsIgnoreCase(originDataType)){
            if (StringUtils.isNotEmpty(value)){
                ps.setDouble(index,Double.valueOf(value.trim()));
            } else {
                ps.setNull(index, Types.DOUBLE);
            }
        } else if ("FLOAT".equalsIgnoreCase(originDataType)){
            if (StringUtils.isNotEmpty(value)){
                ps.setFloat(index,Float.valueOf(value.trim()));
            } else {
                ps.setNull(index, Types.FLOAT);
            }
        } else if ("DECIMAL".equalsIgnoreCase(originDataType)){
            ps.setBigDecimal(index, StringUtils.isEmpty(value) ? null : BigDecimal.valueOf(Double.valueOf(value.trim())));
        } else if ("DATE_TIME".equalsIgnoreCase(originDataType) || "DATE".equalsIgnoreCase(originDataType)){
            if(StringUtils.isNotEmpty(value)){
                Date javaDate = HSSFDateUtil.getJavaDate(Double.valueOf(value.trim()));
                ps.setDate(index, new java.sql.Date(javaDate.getTime()));
            } else {
                ps.setNull(index, Types.DATE);
            }
        }


        else {
            ps.setString(index,StringUtils.isEmpty(value) ? null : value.trim());
        }
    }

    @Override
    public DataVo previewImportData(Map<String, Object> params) {
        Long datasourceId = Long.valueOf(checkNotNull(params.get("datasourceId")).toString());
        Long importId = Long.valueOf(checkNotNull(params.get("id")).toString());
        String tableName = checkNotNull(params.get("tableName")).toString();
        List<String> columns = CollectionUtils.filterAndMap(iDatasourceService.getTableColumns(datasourceId, tableName), ColumnVo::getName, columnVo -> !INNER_FIELDS.contains(columnVo.getName()));
        QueryParam queryParam = new QueryParam();
        StringBuilder ans = new StringBuilder();

        ans.append("select ").append(CollectionUtils.join(columns,Function.identity()))
            .append(" from ").append(tableName).append(" where ")
                .append(INNER_FIELD_IMPORT_ID).append("=").append(importId);
        Object obj = params.get("orderByField");
        if (obj != null && StringUtils.isNotEmpty(obj.toString())){
            ans.append(" order by ").append(obj.toString());
        }
        ans.append(" limit 10 ");
        log.info("previewImportData SQL:{}", ans.toString());
        queryParam.setRawSql(ans.toString());
        DataVo dataVo =  iDatasourceService.execSql(datasourceId, queryParam);
        queryParam = new QueryParam();
        queryParam.setRawSql(String.format("select count(1) as total from %s where %s = %d", tableName, INNER_FIELD_IMPORT_ID,importId));
        DataVo tempDataVo =  iDatasourceService.execSql(datasourceId, queryParam);
        dataVo.setRowSize(Integer.valueOf(tempDataVo.getRows().get(0).get("total").toString()));
        return dataVo;
    }

}
